package cn.fanyc.action.mq.kafka.producer.controller;

import cn.fanyc.action.mq.kafka.producer.entry.RestDomain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.Random;

/**
 * 页面请求控制器
 *
 * @author Fanyc
 * @date 2020-06-28 17:34
 */
@Controller
@RequestMapping("/")
public class IndexController {

    /**
     * 日志记录器
     */
    private final Logger logger = LoggerFactory.getLogger(IndexController.class);

    /**
     * Kafka操作器
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 首页
     */
    @RequestMapping({"", "/", "/index"})
    public String index(ModelMap modelMap) {
        return "index";
    }

    /**
     * 发送消息
     *
     * @param topic   主题
     * @param message 消息内容
     * @param count   循环次数
     * @return 是否发送成功
     */
    @RequestMapping("/send")
    @ResponseBody
    public RestDomain send(String topic, String message, Long count) {
        logger.info("==========开始发送：主题={}，消息={}, 循环次数={}==========", topic, message, count);
        for (int i = 0; i < count; i++) {
            final String formatMessage = message + "-" + i;
            ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, formatMessage);
            send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    logger.error("消息发送失败：{}", throwable.getMessage(), throwable);
                }

                @Override
                public void onSuccess(SendResult<String, String> stringStringSendResult) {
                    logger.info("消息发送成功：分区={}，偏移量={}，消息内容={}"
                            , stringStringSendResult.getRecordMetadata().partition()
                            , stringStringSendResult.getRecordMetadata().offset()
                            , stringStringSendResult.getProducerRecord().value());
                }
            });
            // 添加延时，由于Kafka会缓存批量发送，for循环太快，容易全部放入一个分区，看不出效果
            try {
                Thread.sleep(new Random().nextInt(300));
            } catch (InterruptedException ignored) {
            }
        }
        logger.info("==========发送完成==========");
        return RestDomain.success();
    }
}
